package com.rds.benchmark.thread;

import com.rds.benchmark.command.CommandOperator;
import com.rds.benchmark.config.ArgvsConfig;
import com.rds.benchmark.connector.Connector;
import com.rds.benchmark.operator.TypeOperator;
import org.redisson.api.RedissonClient;

import java.util.Date;
import java.util.concurrent.CountDownLatch;

public class StateThread extends Thread {

    private TypeOperator[] operators;
    private String type;
    private CountDownLatch latch;
    private String threadType;

    public void initParameter(TypeOperator[] operators, String type, CountDownLatch latch, String threadType) {
        this.operators = operators;
        this.type = type;
        this.latch = latch;
        this.threadType = threadType;
    }

    @Override
    public void run() {
        Integer req = ArgvsConfig.requests;
        try {
            //System.out.println("State thread barriers waiting count:" + startBarrier.getNumberWaiting());
            Connector.startBarrier.await();
        } catch (Throwable t) {
            System.out.println("State thread start failed!");
            t.printStackTrace();
            return;
        }

        System.out.println(threadType + " " + type + " testing " + (Connector.connector.type() != null ? "in " + Connector.connector.type() + " model " : "") + "start...");

        long count = 0;
        long size = 0;
        boolean notFirst = false;
        long first = 0;
        long second = 0;
        int cur_temp = req == null ? 0 : req;

        while (latch.getCount() > 0) {
            long begin = System.currentTimeMillis();

            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            int cur = 0;
            for (int i = 0; i < ArgvsConfig.threads; i++) {
                TypeOperator op = operators[i];
                int one = op.resetTimes();
                if (one >= 0) {
                    cur += one;
                }
            }
            if (cur == 0) {
                cur = cur_temp;
            } else {
                cur_temp = cur_temp - cur;
            }
            if (latch.getCount() > 0 || req != null) {
                long duration = (System.currentTimeMillis() - begin);

                second = first;
                first = (long) (cur * 1000.0 / duration);
                System.out.println(new Date() + " -- process " + cur + " times in " + duration + "ms(" + first + "/s)");

                if (second > 0) {
                    if (!notFirst) {
                        // 去掉第一条记录
                        notFirst = true;
                    } else {
                        // second 为了去掉最后1条记录
                        count += second;
                        ++size;
                    }
                }
            }
        }

        double p50 = 0;
        double p90 = 0;
        double p95 = 0;
        double p99 = 0;
        long total = 0;
        double duration = 0;
        for (int i = 0; i < CommandOperator.PXX_MAX; ++i) {
            long[] data = CommandOperator.PXXES[i].get();
            p50 += data[0];
            p90 += data[1];
            p95 += data[2];
            p99 += data[3];
            total += data[4];
            duration += data[5];
        }
        p50 /= CommandOperator.PXX_MAX * 1000;
        p90 /= CommandOperator.PXX_MAX * 1000;
        p95 /= CommandOperator.PXX_MAX * 1000;
        p99 /= CommandOperator.PXX_MAX * 1000;
        if (total > 0) {
            duration /= total * 1000;
        } else {
            duration = 0;
        }

        System.out.println(String.format(threadType + " testing end, average %d/s, latency (AVG: %.3f, P50: %.3f, P90: %.3f, P95: %.3f, P99: %.3f)\n"
                , (size > 0 ? count / size : 0)
                , duration
                , p50, p90, p95, p99
        ));
        Object jedis = Connector.connector.getJedis();
        if (jedis instanceof RedissonClient) {
            ((RedissonClient) jedis).shutdown();
        }
    }
}
